package actor

import (
	"go_learn/mlog"
	"sync/atomic"
)

type ConcurrentActor struct {
	BaseActor
	runNum int32
}

func (slf *ConcurrentActor) OnInit(app App, s Service, addr uint) {
	slf.BaseActor.OnInit(app, s, addr)
	slf.BaseActor.scheduler = slf.scheduler
}

func (slf *ConcurrentActor) Fork(f func(...interface{}), args ...interface{}) {
	sig := ForkSignal{f, args}
	select {
	case slf.lowChan <- &sig:
	default:
		go func() {
			slf.lowChan <- &sig
		}()
	}
}

func (slf *ConcurrentActor) Call(addr interface{}, msgType EMsgType, method string, param ...interface{}) (rst interface{}, err error) {
	rc := make(chan *RetInfo, 0)
	s := slf.newSession(rc, nil)
	msg := RpcMsg{
		Source:  slf.GetAddr(),
		Session: s.id,
		MsgType: msgType,
		Method:  method,
		Params:  param,
		retChan: slf.highChan,
	}
	slf.appInst.Send(addr, &msg)
	slf.resumeMain() //唤醒主协程
	mlog.Debugf("actor %d call %s method %s start", slf.GetAddr(), addr, method)
	atomic.AddInt32(&slf.runNum, -1)
	retMsg := <-rc
	mlog.Debugf("actor %d call %s method %s end", slf.GetAddr(), addr, method)
	atomic.AddInt32(&slf.runNum, 1)
	return retMsg.ret, retMsg.err
}

func (slf *ConcurrentActor) resumeMain() {
	num := atomic.LoadInt32(&slf.runNum)
	if num != 1 {
		mlog.Errorf("run num wrong ", num)
	}
	slf.closeChan <- false
}

func (slf *ConcurrentActor) work(f interface{}, args ...interface{}) {
	defer func() {
		slf.resumeMain()
		atomic.AddInt32(&slf.runNum, -1)
		if r := recover(); r != nil {
			mlog.Errorf("work occur %v", r)
		}
	}()
	switch rf := f.(type) {
	case func(...interface{}):
		rf(args)
	case func(interface{}, error):
		var err error
		if len(args) > 1 {
			err = args[1].(error)
		}
		rf(args[0], err)
	case func(*RpcMsg):
		rf(args[0].(*RpcMsg))
	}
}

func (slf *ConcurrentActor) forkWork(f interface{}, args ...interface{}) {
	atomic.AddInt32(&slf.runNum, 1)
	go slf.work(f, args...)
}

func (slf *ConcurrentActor) dealSignal(sig interface{}) {
	switch msg := sig.(type) {
	case *RpcMsg:
		slf.forkWork(slf.dispatch, msg)
	case *RetInfo:
		s := slf.rmSession(msg.session)
		if s == nil {
			mlog.Errorf("can't find session[%d]", msg.session)
			slf.forkWork(nil)
		} else if s.cb != nil {
			slf.forkWork(s.cb, msg.ret, msg.err)
		} else {
			s.c <- msg
		}
	case *ForkSignal:
		slf.forkWork(msg.f, msg.args...)
	}
}

func (slf *ConcurrentActor) scheduler(closeSig chan bool) {
	defer func() {
		close(slf.closeChan)
		close(slf.lowChan)
		close(slf.highChan)
	}()
	slf.runNum = 1
	for {
		var sig interface{}
		select {
		case sig = <-slf.highChan:
		default:
			select {
			case sig = <-slf.highChan:
			case sig = <-slf.lowChan:
			}
		}
		slf.dealSignal(sig)
		atomic.AddInt32(&slf.runNum, -1)
		exit := <-slf.closeChan //休眠,等待被唤醒
		atomic.AddInt32(&slf.runNum, 1)
		if exit {
			return
		}
	}
}
